package com.shujia.jinjie;

import com.shujia.utils.HBaseUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.Scanner;

/**
 * 1、通过手机号查询用户最近10条位置记录
 * 2、获取用户某一天在一个城市中的所有位置
 */
public class DianXinRowKeyDemo {
    private static Connection conn = HBaseUtil.CONNECTION;
    private static Admin admin = HBaseUtil.ADMIN;


    public static void main(String[] args) throws Exception {
        //读取文件将数据添加到hbase表中
//        putsData();

        //1、通过手机号查询用户最近10条位置记录
        Scanner sc = new Scanner(System.in);
        System.out.println("请输入要查询的手机号：");
        String phoneNum = sc.next();

//        findUser(phoneNum);

        //2、获取用户某一天在一个城市中的所有位置
        System.out.println("请输入要查询的城市编号：");
        String city = sc.next();
        findUserPosition(phoneNum,city);


    }

    public static void findUserPosition(String number,String city) throws Exception{
        //获取表对象
        Table dianxinTb1 = conn.getTable(TableName.valueOf("dianxin_tb1"));

        //使用行键前缀过滤器
        Scan scan = new Scan();

        //过滤手机号行键
        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(number));

        //过滤城市
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("city"),
                CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes(city)));

        //创建组合过滤器对象
        FilterList filterList = new FilterList();
        filterList.addFilter(prefixFilter);
        filterList.addFilter(singleColumnValueFilter);

        scan.setFilter(filterList);
        ResultScanner resultScanner = dianxinTb1.getScanner(scan);
        for (Result result : resultScanner) {
            String rk = Bytes.toString(result.getRow());
            //47BE1E866CFC071DB19D5E1C056BE28AE24C16E7-9496768070
            String[] strings = rk.split("-");
            String phoneNum = strings[0];
            long startTime = 20190000000000L - Long.parseLong(strings[1]);

            String wg = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("wg")));
            String city2 = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("city")));
            String qx = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("qx")));
            String stayTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("stayTime")));
            String endTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("endTime")));
            String date = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("date")));

            System.out.println("手机号:"+phoneNum+"，进入网格时间:"+startTime+"，离开网格时间:"+endTime
                    +"，网格编号:"+wg+"，城市:"
                    +city2+"，区县:"+qx+"，停留时间:"
                    +stayTime+"，日期:"+date);

        }
    }

    public static void findUser(String number) throws Exception{
        //获取表对象
        Table dianxinTb1 = conn.getTable(TableName.valueOf("dianxin_tb1"));

        //使用行键前缀过滤器
        Scan scan = new Scan();

        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(number));

        scan.setFilter(prefixFilter);
        scan.setLimit(10);
        ResultScanner resultScanner = dianxinTb1.getScanner(scan);
        for (Result result : resultScanner) {

            String rk = Bytes.toString(result.getRow());
            //47BE1E866CFC071DB19D5E1C056BE28AE24C16E7-9496768070
            String[] strings = rk.split("-");
            String phoneNum = strings[0];
            long startTime = 20190000000000L - Long.parseLong(strings[1]);


            String wg = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("wg")));
            String city = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("city")));
            String qx = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("qx")));
            String stayTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("stayTime")));
            String endTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("endTime")));
            String date = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("date")));

            System.out.println("手机号:"+phoneNum+"，进入网格时间:"+startTime+"，离开网格时间:"+endTime
                    +"，网格编号:"+wg+"，城市:"
                    +city+"，区县:"+qx+"，停留时间:"
                    +stayTime+"，日期:"+date);

        }
    }

    public static void putsData() throws Exception {


        //获取表对象
        Table dianxinTb1 = conn.getTable(TableName.valueOf("dianxin_tb1"));

        BufferedReader br = new BufferedReader(new FileReader("hbase/data/dianxin_data"));

        String line = null;
        while ((line = br.readLine()) != null) {
            if (!(line.contains("shushan")) && !(line.contains("\\N"))) {
                String[] strings = line.split("\t");
                String phoneNum = strings[0];
                String wg = strings[1];
                String city = strings[2];
                String qx = strings[3];
                String stayTime = strings[4];
                String startTime = strings[5];
                String endTime = strings[6];
                String date = strings[7];

                //我们为了需求的查询更加方便，以手机号与进入网格的时间进行拼接
                //因为要查询最近的，可以使用时间戳反转对rowkey进行设计
                //20180503173254
                //20190000000000
                long new_time = 20190000000000L - Long.parseLong(startTime);
                String rk = phoneNum + "-" + new_time;
                HBaseUtil.putOneDataToTable("dianxin_tb1", rk, "info", "wg", wg);
                HBaseUtil.putOneDataToTable("dianxin_tb1", rk, "info", "city", city);
                HBaseUtil.putOneDataToTable("dianxin_tb1", rk, "info", "qx", qx);
                HBaseUtil.putOneDataToTable("dianxin_tb1", rk, "info", "stayTime", stayTime);
                HBaseUtil.putOneDataToTable("dianxin_tb1", rk, "info", "endTime", endTime);
                HBaseUtil.putOneDataToTable("dianxin_tb1", rk, "info", "date", date);
            }
        }
    }
}
